package com.lzwk.utils;

/**
 * @Author: CC
 * @Date: 2022/1/5 17:57
 */

import com.alibaba.fastjson.JSON;
import com.lzwk.common.ocpx.OCPXConstant;
import com.lzwk.common.test.TestConstant;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;


import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;
import java.util.Properties;


public class MyKafkaUtil {
    private static String brokers = "127.0.0.1:9092";
    private static String default_topic = "DWD_DEFAULT_TOPIC";

    public static FlinkKafkaProducer<String> getKafkaProducer(String topic,String brokers) {
        return new FlinkKafkaProducer<String>(brokers,
                topic,
                new SimpleStringSchema());
    }

    public static <T> FlinkKafkaProducer<T> getKafkaProducer(KafkaSerializationSchema<T> kafkaSerializationSchema) {

        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);

        return new FlinkKafkaProducer<T>(default_topic,
                kafkaSerializationSchema,
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }


    public static <T> FlinkKafkaProducer<T> getKafkaProducer(String topic,String flag,String broker) {

        if("test".equals(flag) ){
            Properties properties = new Properties();
            properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
/*            properties.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "900000");
            properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
            properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "32768");
            properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "5");
            properties.setProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "3145728");*/

            return new FlinkKafkaProducer<T>(default_topic,
                    new KafkaSerializationSchema<T>() {
                        @Override
                        public ProducerRecord<byte[], byte[]> serialize(T t, @Nullable Long aLong) {
                            return new ProducerRecord<>(topic, JSON.toJSONString(t).getBytes(StandardCharsets.UTF_8));
                        }
                    },
                    properties,
                    FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        }else{
            Properties properties = new Properties();
            properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
            properties.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, "900000");
            properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
            properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "32768");
            properties.setProperty(ProducerConfig.LINGER_MS_CONFIG, "5");
            properties.setProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "3145728");

            return new FlinkKafkaProducer<T>(default_topic,
                    new KafkaSerializationSchema<T>() {
                        @Override
                        public ProducerRecord<byte[], byte[]> serialize(T t, @Nullable Long aLong) {
                            return new ProducerRecord<>(topic, JSON.toJSONString(t).getBytes(StandardCharsets.UTF_8));
                        }
                    },
                    properties,
                    FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
        }

    }

    public static FlinkKafkaConsumer<String> getKafkaConsumer(String topic, String groupId,String brokers) {

        Properties properties = new Properties();

        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        return new FlinkKafkaConsumer<String>(topic,
                new SimpleStringSchema(),
                properties);

    }

    //拼接Kafka相关属性到DDL(source)
    public static String getKafkaDDL(String topic, String groupId) {
        return  " 'connector' = 'kafka', " +
                " 'topic' = '" + topic + "'," +
                " 'properties.bootstrap.servers' = '" + brokers + "', " +
                " 'properties.group.id' = '" + groupId + "', " +
                " 'format' = 'json', " +
                " 'scan.startup.mode' = 'latest-offset'  ";
    }
    //拼接Kafka相关属性到DDL(sink)
    public static String getKafkaDDL(String topic) {
        return  " 'connector' = 'upsert-kafka', " +
                " 'topic' = '" + topic + "'," +
                " 'properties.bootstrap.servers' = '" + brokers + "', " +
                " 'key.format' = 'json', " +
                " 'key.json.ignore-parse-errors' = 'true', " +
                " 'value.format' = 'json', "+
                " 'value.json.fail-on-missing-field' = 'false' ";
    }
}
